package com.liuguang.rabbitmq.comsumer;

import com.liuguang.common.constans.CommonConstans;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: TestListener
 * @Description：RabbitMq测试
 * @Author: wanghr
 * Date: 1:21 PM 2020/10/16
 * Version: 1.0
 */
@Component
public class TestListener {
    private static final Logger logger = LoggerFactory.getLogger(TestListener.class);

    /**
     * 第一个，监听测试队列的消费者
     * 用来测试自动ack
     */
//    @RabbitListener(queues = CommonConstans.TEST_QUEUE,errorHandler = "myMqExceptionHandler")
    public void testListener1(Message message, Channel channel) throws IOException {
        logger.info(new String(message.getBody()));
        MessageProperties properties = message.getMessageProperties();

//        try {
//            TimeUnit.SECONDS.sleep(10);
//            TimeUnit.HOURS.sleep(10);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

        int a = 1 / 0;

    }

    /**
     * 第二个，监听测试队列的消费者
     * 用来进行手动 ack 的测试
     */
    @RabbitListener(queues = CommonConstans.TEST_QUEUE)
    public void testListener2(Message message, Channel channel) throws IOException {
        logger.info("第二个消费者收到测试消息");
        String s = new String(message.getBody());
        if ("第200条消息".equals(s)) {
            try {
                int a = 1 / 0;
            } catch (Exception e) {
                logger.info("模拟出现异常，手动拒绝,{}",s);
                //第三个参数对应是否将未处理成功的消息放回队列
                // 在只有一个消费者的情况下，如果对应的消息出现异常，会阻塞后面的消息意味着后面的消息也不能被消费
//                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
                //将第三个参数设为false，消息丢弃或者放入死信就不会产生，上述问题
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
                return;
//                SimpleMessageListenerContainer
//                ScheduledExecutorService
            }
        }
        logger.info("{},处理成功",s);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    //测试：
    // 1、在自动提交状态下，当第一个消费者抛出异常，会如何执行？
    //      消息向下传递给另一个消费者，如果只有一个消费者，那么将不断的向该消费者发送消息，抛异常
    // 2、是否消息到broker后就进行回调，与消息消费无关?         方法：第一个线程睡眠，查看是否回调方法即时执行？或者直接就没有消费者去消费消息
    //  到后就回调，与消费无关
    // 3、当消息没有消费者，会一直存在队列中，直到有消费者。
    // 4、在手动提交模式下，没有手动ack，出现什么情况？
    // 当重启前队列中 一直保存消息，并且是未ack 状态，当服务关闭后，这条消息会变为准备发送状态，服务重启再次发送消息。
    // 即使有多个消费者，当消息被分发到某个，却没有ack，那么这个消息一直处于未ack状态
    // 5、只有一个消费者，处理的时间很长，消息长生很快，会堆积在队列吗？如果此条消息产生异常会一直重试，会阻塞后面的消息吗？
    // 当达到消费者配置的prefetch数量，消息会在队列中堆积，
    // 当消息产生异常，不会阻塞，但会根据ack方式决定是否不断重试：自动方式，不断重试；手动会执行一次
    //todo: 消息的持久化
    //      私信队列的使用，解决未处理消息
}
